#ifndef _RELATION_MANAGER_HPP_
#define _RELATION_MANAGER_HPP_

#include <cstring>
#include <cstdlib>
#include <cstdio>
#include <ctime>
#include <vector>
#include <stdint.h>
#include "ThreadCounter.hpp"

#define KB 1024
#define MB (KB * KB)

struct RelationPair
{
    uint32_t id1;
    uint32_t id2;
    uint32_t weight;
    static int Comp(const void *a, const void *b)
    {
        const RelationPair *pa = reinterpret_cast<const RelationPair*>(a);
        const RelationPair *pb = reinterpret_cast<const RelationPair*>(b);

        if (pa->id1 < pb->id1)
            return -1;
        else if (pa->id1 > pb->id1)
            return 1;
        else {
            if (pa->id2 < pb->id2)
                return -1;
            else if (pa->id2 > pb->id2)
                return 1;
            else
                return 0;
        }
    }
    bool operator<(const RelationPair& b) const {
        if (id1 < b.id1)
            return true;
        else if (id1 == b.id1 && id2 < b.id2)
            return true;
        return false;
    }

    bool operator==(const RelationPair& b) const {
        return id1 == b.id1 && id2 == b.id2;
    }

};


class RelationBucket 
{
    static const uint32_t BUF_BYTES_LIMIT = 64 * MB;
    RelationPair *_pairs;
    uint32_t _limit;
    uint32_t _size;
    uint32_t _myid;
    const char *_file_path;

    static void *SortAndDump(void *instance)
    {
        RelationBucket *bucket = reinterpret_cast<RelationBucket*>(instance);
        char file_name[256];
        uint32_t now = time(NULL);

        snprintf(file_name, 256, "%s/rel.%04u.%u", bucket->_file_path, bucket->_myid, now);

        printf("Begin to dump&sort buckets %u\n", bucket->_myid);

        qsort(bucket->_pairs, bucket->_size, sizeof(RelationPair), RelationPair::Comp);

        
        RelationPair *sorted_pairs = new RelationPair[bucket->_size];
        uint32_t sorted_size = 0;

        /* merge(deduplicate) */
        RelationPair base;
        base.id1 = 0xffffffff;
        base.id2 = 0xffffffff;

        for(uint32_t i = 0; i < bucket->_size; i++) {
            if (bucket->_pairs[i].id1 != base.id1 
                || bucket->_pairs[i].id2 != base.id2) {
                if (i) {
                    sorted_pairs[sorted_size] = base;
                    sorted_size++;
                }

                base = bucket->_pairs[i];
            } else {
                /* merge */
                base.weight += bucket->_pairs[i].weight;
            }
        }
        
        FILE *f = fopen(file_name, "wb");
        fwrite(sorted_pairs, sizeof(RelationPair), sorted_size, f);
        fclose(f);
        printf("dump&sort buckets %u end, sortmerged:raw(%u:%u)\n",
                bucket->_myid, sorted_size, bucket->_size);
        delete bucket;
        delete sorted_pairs;
        ThreadCounter::EndThread();
    }


public:
    RelationBucket(const char *file_path, uint32_t id) 
        : _limit(BUF_BYTES_LIMIT / sizeof(RelationPair)),
        _size(0),
        _file_path(file_path),
        _myid(id)
    {
        _pairs = new RelationPair[_limit];
    }

    RelationBucket(RelationBucket &prev)
        :_pairs(prev._pairs),
        _limit(prev._limit),
        _size(prev._size),
        _file_path(prev._file_path),
        _myid(prev._myid)
    {
        prev._size = 0;
        prev._pairs = new RelationPair[_limit];
    }

    ~RelationBucket()
    {
        delete[] _pairs;
    }


    void Dump()
    {
        RelationBucket *dump_bucket = new RelationBucket(*this);
        pthread_t tid;
        ThreadCounter::BeginThread();
        if (pthread_create(&tid, NULL, RelationBucket::SortAndDump, dump_bucket) < 0) {
            printf("ERROR: create thread failed\n");
            exit(-1);
        }
        pthread_detach(tid);
    }

    void Add(uint32_t id1, uint32_t id2, uint32_t weight)
    {
        _pairs[_size].id1 = id1;
        _pairs[_size].id2 = id2;
        _pairs[_size].weight = weight;
        _size++;
        if (_size == _limit) {
            Dump();
        }
    }

};



class RelationManager
{
    const char *_file_path;
    uint32_t _bucket_num;
    RelationBucket **_buckets;


public:
    RelationManager(const char *file_path, uint32_t bucket_num)
        : _file_path(file_path), _bucket_num(bucket_num)
    {
        _buckets = new RelationBucket*[_bucket_num];
        for(uint32_t i = 0; i < _bucket_num; i++) 
            _buckets[i] = new RelationBucket(_file_path, i);
    }


    ~RelationManager()
    {
        for(uint32_t i = 0; i < _bucket_num; i++) 
            delete _buckets[i];
        delete []_buckets;
    }

    
    void AddRelation(uint32_t id1, uint32_t id2, uint32_t weight)
    {
        uint32_t bucket_id = (id1 * 21788233) % _bucket_num;
        _buckets[bucket_id]->Add(id1, id2, weight);
    }

    void Done()
    {
        for(uint32_t i = 0; i < _bucket_num; i++) {
            _buckets[i]->Dump();
        }
    }

    /*
    void Shrink() 
    {
        struct dirent **filelist;

        int fnum = scandir(_file_path, &filelist, 0, alphasort);

        int last_bid = 0;
        std::vector<RelationPair>
        
        for(int i = 0; i < fnum; i++) {
            const char *cstr = filelist[i]->d_name;
            if (strncmp(cstr, "rel.", 4))
                continue;

            int cur_bid = atoi(cstr + 4);
            if (cur_bid != last_bid) {
                distance_manager.DumpResult(book_name);
                distance_manager.ClearData();
                last_bid = cur_bid;
            }

            

        }



        for(uint32_t i = 0; i < _bucket_num; i++)
        RelationBucket *bucket = reinterpret_cast<RelationBucket*>(instance);
        char file_name[256];
        uint32_t now = time(NULL);

        snprintf(file_name, 256, "%s/rel.%04u.%u", bucket->_file_path, bucket->_myid, now);

        printf("Begin to dump&sort buckets %u\n", bucket->_myid);

        qsort(bucket->_pairs, bucket->_size, sizeof(RelationPair), RelationPair::Comp);

        
        RelationPair *sorted_pairs = new RelationPair[bucket->_size];
        uint32_t sorted_size = 0;

        RelationPair base;
        base.id1 = 0xffffffff;
        base.id2 = 0xffffffff;

        for(uint32_t i = 0; i < bucket->_size; i++) {
            if (bucket->_pairs[i].id1 != base.id1 
                || bucket->_pairs[i].id2 != base.id2) {
                if (i) {
                    sorted_pairs[sorted_size] = base;
                    sorted_size++;
                }

                base = bucket->_pairs[i];
            } else {
                base.weight += bucket->_pairs[i].weight;
            }
        }
        
        FILE *f = fopen(file_name, "wb");
        fwrite(sorted_pairs, sizeof(RelationPair), sorted_size, f);
        fclose(f);
        printf("dump&sort buckets %u end, sortmerged:raw(%u:%u)\n",
                bucket->_myid, sorted_size, bucket->_size);
        delete bucket;
        delete sorted_pairs;
        ThreadCounter::EndThread();

    }
    */
};


#endif

